#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
#include <assert.h>
#include "queue.h"
#include "pool.h"


static void *thread_function(void *arg);

struct threadpool *create_threadpool(int queue_max_length,int count)
{
	struct threadpool *pool = NULL;
	int ret = 0;

	pool = malloc(sizeof(struct threadpool));
	if(pool == NULL)
	{
		printf("malloc error\n");
		return NULL;
	}
	memset(pool,0,sizeof(struct threadpool));

	pool->queue_max_num = queue_max_length;
	pool->que = create_queue();

	if(sem_init(&(pool->semr),0,0) == -1)
	{
		perror("sem_init");
		return NULL;
	}
	if(sem_init(&(pool->semw),0,5) == -1)
	{
		perror("sem_init");
		return NULL;
	}


	if(pthread_mutex_init(&(pool->mutex),NULL) == -1)
	{
		printf("init mutex failed\n");
		return NULL;
	}

	pool->tids = malloc(sizeof(pthread_t) * count);
	if(pool->tids == NULL)
	{
		printf("malloc error\n");
		return NULL;
	}
	memset(pool->tids,0,sizeof(pthread_t) * count);
	pool->count = count;

	for(int i = 0;i < count;i++)
	{
		ret = pthread_create(&pool->tids[i],NULL,thread_function,pool);
		if(ret != 0)
		{
			printf("pthread create error\n");
			exit(-1);
		}
	}

	return pool;
}


void *thread_function(void *arg)
{
	struct threadpool *pool = (struct threadpool *)arg;
	struct task task;
	memset(&task,0,sizeof(struct task));

	while(1)
	{
		sem_wait(&pool->semr);
		pthread_mutex_lock(&pool->mutex);
		task = front(pool->que);
		pop(pool->que);
		sem_post(&pool->semw);
		pthread_mutex_unlock(&pool->mutex);

		task.callback_function(task.arg);
		free(task.arg);
	}

}

int threadpool_destroy(struct threadpool *pool)
{
	int ret = 0;
	assert(pool != NULL);

	for(int i = 0;i < pool->count;i++)
	{
		ret = pthread_join(pool->tids[i],NULL);
		if(ret == -1)
		{
			perror("pthread_join");
			exit(-1);
		}
	}

	pthread_mutex_destroy(&pool->mutex);
	sem_destroy(&pool->semr);
	sem_destroy(&pool->semw);

	free(pool->tids);
	free(pool);

	return 0;
}


int threadpool_add_task(struct threadpool *pool,CALLBACK callback_function,void *arg)
{

	struct task task;

	memset(&task,0,sizeof(struct task));
	task.callback_function = callback_function;
	task.arg = arg;

	sem_wait(&pool->semw);
	pthread_mutex_lock(&pool->mutex);
	push(pool->que,task);
	pthread_mutex_unlock(&pool->mutex);
	sem_post(&pool->semr);

	return 0;
}






